package cn.young.dev.stock;



import cn.young.dev.util.KafkaProducer;
import cn.young.dev.util.MysqlConnector;
import kafka.utils.Json;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * @Auther: Ryan
 * @Date: 2022/1/27 16:03
 * @Description:
 */
public class StockStreamProducer extends Thread {

    public static final String STOCK_DAILY_PREFIX = "stock_daily_";

    public static final String dbUrl = "jdbc:mysql://www.ryans.wiki:3306/test";
    public static final String userName = "root";
    public static final String password = "123456";

    public static final String ZOOKEEPER_CONNECT = "192.168.16.1:2181,192.168.16.1:2182,192.168.16.1:2183";
    public static final String KAFKA_BROKER_LIST = "192.168.16.1:9092,192.168.16.1:9093,192.168.16.1:9094";

    private String code;

    public StockStreamProducer(String code) {
        this.code = code;
    }

    @Override
    public void run() {
        try {
            MysqlConnector connector = new MysqlConnector(dbUrl, userName, password);
            String sql = "select ts_code from stock_daily where ts_code='" + code + "'  order by trade_date ASC";
            List<Map<String, String>> searchList = connector.search(sql, Arrays.asList("tsCode", "tradeDate", "open", "high", "low", "close", "pre_close"));
            for(Map<String, String> map : searchList){
                new KafkaProducer(ZOOKEEPER_CONNECT, KAFKA_BROKER_LIST).send(STOCK_DAILY_PREFIX + code, Json.encode(map));
                Thread.sleep(100);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
